package com.zwh.rabbitmq.config;

import java.text.MessageFormat;
import java.time.Duration;
import java.util.Optional;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.boot.context.properties.PropertyMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.RetryListener;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;

import com.zwh.rabbitmq.config.aspect.MessageProcess;
import com.zwh.rabbitmq.config.dto.RabbitmqMessageRecordDto;


/**
 * rabbitmq配置类
 * @author zwh
 * @date 2021/12/1 15:15
 **/
@Configuration
public class RabbitMqConfig {
    Logger logger = LoggerFactory.getLogger(RabbitMqConfig.class);

    private final ObjectProvider<MessageConverter> messageConverter;

    private final RabbitProperties properties;

    public RabbitMqConfig(
            ObjectProvider<MessageConverter> messageConverter,
            RabbitProperties properties) {
        this.messageConverter = messageConverter;
        this.properties = properties;
    }
    /**
     * 设置autoConfiguration中的默认配置bean
     * @param connectionFactory
     * @return
     */
//    @Bean
//    @Primary
//    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
//        PropertyMapper map = PropertyMapper.get();
//        RabbitTemplate template = new RabbitTemplate(connectionFactory);
//        MessageConverter messageConverter = this.messageConverter.getIfUnique();
//        if (messageConverter != null) {
//            template.setMessageConverter(messageConverter);
//        }
//        template.setMandatory(determineMandatoryFlag());
//
//        RabbitProperties.Template properties = this.properties.getTemplate();
//        if (properties.getRetry().isEnabled()) {
//            template.setRetryTemplate(createRetryTemplate(properties.getRetry()));
//        }
//        map.from(properties::getReceiveTimeout).whenNonNull().as(Duration::toMillis)
//                .to(template::setReceiveTimeout);
//        map.from(properties::getReplyTimeout).whenNonNull().as(Duration::toMillis)
//                .to(template::setReplyTimeout);
//        map.from(properties::getExchange).to(template::setExchange);
//        map.from(properties::getRoutingKey).to(template::setRoutingKey);
//        return template;
//    }
//    @Bean
//    public Jackson2JsonMessageConverter producerJackson2MessageConverter() {
//        return new Jackson2JsonMessageConverter();
//    }
    /**
     * 自定义异步确认rabbitmq访问bean
     * @param connectionFactory
     * @param messageProcess
     * @return
     */
    @Bean("rabbitJsonTemplate")
    public ReliableRabbitTemplate rabbitJsonTemplate(ConnectionFactory connectionFactory,
                                                     MessageProcess messageProcess) {
        ReliableRabbitTemplate template = new ReliableRabbitTemplate(messageProcess);
        template.setConnectionFactory(connectionFactory);

        template.setMessageConverter(new Jackson2JsonMessageConverter());

        // 异步,消息回退回调 ReturnCallback
        template.setMandatory(true);

        RetryTemplate retryTemplate = new RetryTemplate();

        retryTemplate.setRetryPolicy(new SimpleRetryPolicy(5));
        // retryTemplate.setRetryPolicy(Optional.ofNullable(messageProcess.getRetryPolicy()).orElse(policy));
        retryTemplate.registerListener(new RetryListener() {
            @Override
            public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
                // 开始调用
                System.out.println("开始调用");
                return true;
            }

            @Override
            public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback, Throwable throwable) {
                // 结束调用
                System.out.println("结束调用");
            }

            @Override
            public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback, Throwable throwable) {
                // 每次重试失败后都会调用
                System.out.println("每次重试失败后都会调用：次数：" + context.getRetryCount());
            }
        });

        template.setRetryTemplate(retryTemplate);

        //确认回调
        template.setConfirmCallback((correlationData, ack, cause) -> {
            RabbitmqMessageRecordDto messageInfo = null;
            if (correlationData != null && correlationData instanceof ExtCorrelationData) {
                ExtCorrelationData extData = (ExtCorrelationData) correlationData;
                messageInfo = extData.getMessageInfo();
            }
//          logger.info("rabbitmq服务发布确认消息：ack:{}, cause:{}", ack, cause);

            if (ack) {
                // ack成功，MQ已成功收到消息 （路由失败时也会正常响应）
                messageProcess.getMessageAck().accept(messageInfo);
            } else {
                // 发送消息失败，MQ异常或网络异常
                messageInfo.setDescription(Optional.ofNullable(cause).orElse(""));
                messageProcess.getMessageNack().accept(messageInfo);
            }
        });

        //找不到路由回调
        template.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
            String messageBody = template.getBodyStr(message);
            String messageProperties = template.getPropertiesStr(message);
            String msg = MessageFormat.format("replyCode:{0}, replyText:{1}", replyCode, replyText);
            messageProcess.getMessageRouteFail().accept(message, messageProperties, messageBody, exchange, routingKey, msg);
        });
        return template;
    }


    private boolean determineMandatoryFlag() {
        Boolean mandatory = this.properties.getTemplate().getMandatory();
        return (mandatory != null) ? mandatory : this.properties.isPublisherReturns();
    }

    private RetryTemplate createRetryTemplate(RabbitProperties.Retry properties) {
        PropertyMapper map = PropertyMapper.get();
        RetryTemplate template = new RetryTemplate();
        SimpleRetryPolicy policy = new SimpleRetryPolicy();
        map.from(properties::getMaxAttempts).to(policy::setMaxAttempts);
        template.setRetryPolicy(policy);
        ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
        map.from(properties::getInitialInterval).whenNonNull().as(Duration::toMillis)
                .to(backOffPolicy::setInitialInterval);
        map.from(properties::getMultiplier).to(backOffPolicy::setMultiplier);
        map.from(properties::getMaxInterval).whenNonNull().as(Duration::toMillis)
                .to(backOffPolicy::setMaxInterval);
        template.setBackOffPolicy(backOffPolicy);
        return template;
    }
}
